/*
 * Licensed to the Deppon EXPRESS .
 */
package com.depponframework.rocketmq.spring.autoconfigure;

import com.depponframework.rocketmq.spring.annotation.ConsumeMode;
import com.depponframework.rocketmq.spring.annotation.MessageModel;
import com.depponframework.rocketmq.spring.annotation.RocketMessageListener;
import com.depponframework.rocketmq.spring.core.RocketConsumerWrapper;
import com.depponframework.rocketmq.spring.core.RocketListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinitionCustomizer;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.StandardEnvironment;

import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Configuration
public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {

    private ConfigurableApplicationContext applicationContext;
    private AtomicLong counter = new AtomicLong(0L);
    private StandardEnvironment environment;
    private RocketProperties rocketProperties;

    @Autowired
    public ListenerContainerConfiguration(StandardEnvironment environment, RocketProperties rocketProperties) {
        this.environment = environment;
        this.rocketProperties = rocketProperties;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

    @Override
    public void afterSingletonsInstantiated() {
        Map<String, Object> listenerBeans = this.applicationContext.getBeansWithAnnotation(RocketMessageListener.class);
        if (Objects.nonNull(listenerBeans)) {
            listenerBeans.forEach(this::registerContainer);
        }
    }

    private void registerContainer(String listenerBeanName, Object listenerBean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(listenerBean);
        if (!RocketListener.class.isAssignableFrom(listenerBean.getClass())) {
            throw new IllegalStateException(clazz + " is not instance of " + RocketListener.class.getName());
        } else {
            RocketMessageListener rocketListenerAnnotation = clazz.getAnnotation(RocketMessageListener.class);
            this.validate(rocketListenerAnnotation);

            String consumerName = String.format("%s_%s", RocketConsumerWrapper.class.getName(), this.counter.incrementAndGet());
            GenericApplicationContext genericApplicationContext = (GenericApplicationContext) this.applicationContext;
            genericApplicationContext.registerBean(
                    consumerName,
                    RocketConsumerWrapper.class, () -> this.createRocketConsumerWrapper(consumerName, listenerBean, rocketListenerAnnotation),
                    new BeanDefinitionCustomizer[0]);

            RocketConsumerWrapper rocketConsumer = genericApplicationContext.getBean(consumerName, RocketConsumerWrapper.class);
            if (!rocketConsumer.isRunning()) {
                try {
                    rocketConsumer.start();
                } catch (Exception e) {
                    log.error("Started container failed. {}", rocketConsumer, e);
                    throw new RuntimeException(e);
                }
            }

            log.info("Register the listener to container, listenerBeanName:{}, consumerName:{}", listenerBeanName, consumerName);
        }
    }

    private void validate(RocketMessageListener annotation) {
        if (annotation.consumeMode() == ConsumeMode.ORDERLY && annotation.messageModel() == MessageModel.BROADCASTING) {
            throw new BeanDefinitionValidationException("Bad annotation definition in @RocketMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
        }
    }

    private RocketConsumerWrapper createRocketConsumerWrapper(String consumerName, Object listenerBean, RocketMessageListener rocketListenerAnnotation) {
        RocketConsumerWrapper consumerWrapper = new RocketConsumerWrapper();

        RocketProperties rocketPropertiesCopy = new RocketProperties();
        rocketPropertiesCopy.setAccessChannel(rocketListenerAnnotation.accessChannel());

        try {
            BeanUtils.copyProperties(rocketPropertiesCopy, this.rocketProperties);
        } catch (IllegalAccessException | InvocationTargetException ignored) {
        }

        RocketProperties.Consumer consumerPropertiesCopy = rocketPropertiesCopy.getConsumer();
        consumerPropertiesCopy.setConsumerGroup(rocketListenerAnnotation.consumerGroup());
        consumerPropertiesCopy.setTopic(rocketListenerAnnotation.topic());
        consumerPropertiesCopy.setConsumeThreadMax(rocketListenerAnnotation.consumeThreadMax());
        consumerPropertiesCopy.setConsumeTimeout(rocketListenerAnnotation.consumeTimeout());
        consumerPropertiesCopy.setAccessKey(rocketListenerAnnotation.accessKey());
        consumerPropertiesCopy.setSecretKey(rocketListenerAnnotation.secretKey());
        consumerPropertiesCopy.setEnableMsgTrace(rocketListenerAnnotation.enableMsgTrace());
        consumerPropertiesCopy.setNameServerAddress(rocketListenerAnnotation.nameServer());

        consumerWrapper.setConsumerProperties(consumerPropertiesCopy);
        consumerWrapper.setRocketListener((RocketListener) listenerBean);
        consumerWrapper.setConsumerName(consumerName);

        return consumerWrapper;
    }

}
